/*
 * Copyright (c) 2022, Chongqing UCchip InfoTech Co.,Ltd
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * Change Logs:
 * Date           Author       Notes
 * 2023-12-18     jlu          the first version
 */
#include <rtthread.h>
#ifdef PKG_USING_MYMQTT
#include "net_mqtt.h"
#include "led_manager.h"
#include "manager_queue.h"
#include "manager_queue_comm.h"
#include "manager_task.h"
#include "module_queue_struct.h"
#include "net_app.h"
#include "state_manager.h"
#include <rthw.h>
#include <config_api.h>
#include <mqtt_client.h>
#include <string.h>

// #define DBG_LVL LOG_LVL_DBG
#define DBG_ENABLE
#define DBG_SECTION_NAME "net"
#include "uc_log.h"

/**
 * @brief  MQTT 客户端
 *
 */
static struct mqtt_client g_mqtt_client = {0};

/**
 * @brief  MQTT 配置
 *
 */
static net_mqtt_config_t g_mqtt_config = {0};

/**
 * @brief  MQTT 启动状态
 *
 */
static int g_mqtt_started = 0;

/**
 * @brief  MQTT 状态
 *
 */
static mqtt_state_e g_mqtt_state = MQTT_STATE_OFFLINE;

/**
 * @brief  MQTT 发布计数器
 *
 */
static unsigned int g_mqtt_publish_num = 1;

/**
 * @brief  MQTT 消息队列
 *
 */
static void *g_net_mqtt_queue = RT_NULL;

#ifdef MQTT_PACKET_GROUP
/**
 * @brief MQTT组包队列
 *
 */
static void *g_mqtt_com_queue = RT_NULL;

/**
 * @brief the MQTT mutex
 *
 */
// static rt_mutex_t g_mqtt_mutex = RT_NULL;
#endif

/**
 * @brief  MQTT 连接回调函数
 *
 * @param  c MQTT 客户端
 */
static void mqtt_connect_callback(mqtt_client *c)
{
}

/**
 * @brief  MQTT 上线回调函数
 *
 * @param  c MQTT 客户端
 */
static void mqtt_online_callback(mqtt_client *c)
{
    g_mqtt_state = MQTT_STATE_ONLINE;
    // send message to queue
    void *queue = net_app_get_queue();

    extern unsigned int g_mqtt_offline_time;
    g_mqtt_offline_time = 0;

    manager_set_mqtt_state(g_mqtt_state);

    if (RT_NULL != queue)
    {
        if (module_queue_communication(queue, GATEWAY_MODULE_DEFAULT, NET_APP_MODULE, NET_MQTT_STATE_CHANGED, (void *)g_mqtt_state))
        {
            TRACE_E("mqtt state changed queue communication failed, mqtt %s", g_mqtt_state ? "online" : "offline");
        }
    }
    else
    {
        TRACE_E("net_app_get_queue null");
    }

    // MQTT 状态正常
    send_led_event(LED_MQTT, LED_E_NORMAL);
}

/**
 * @brief  MQTT 掉线回调函数
 *
 * @param  c MQTT 客户端
 */
static void mqtt_offline_callback(mqtt_client *c)
{
    g_mqtt_state = MQTT_STATE_OFFLINE;

    extern unsigned int g_mqtt_offline_time;
    if (0 == g_mqtt_offline_time)
    {
        g_mqtt_offline_time = rt_tick_get();
    }
    // send message to queue
    void *queue = net_app_get_queue();

    manager_set_mqtt_state(g_mqtt_state);

    if (RT_NULL != queue)
    {
        if (module_queue_communication(queue, GATEWAY_MODULE_DEFAULT, NET_APP_MODULE, NET_MQTT_STATE_CHANGED, (void *)g_mqtt_state))
        {
            TRACE_E("mqtt state changed queue communication failed, mqtt %s", g_mqtt_state ? "online" : "offline");
        }
    }
    else
    {
        TRACE_E("net_app_get_queue null");
    }

    // MQTT 状态异常
    send_led_event(LED_MQTT, LED_E_EXCEPTION);
}

/**
 * @brief  MQTT 订阅消息回调函数
 *
 * @param  c 客户端
 * @param  msg_data 订阅消息
 */
static void mqtt_sub_callback(mqtt_client *c, message_data *msg_data)
{
    // 创建 MQTT 消息
    net_mqtt_message_t *mqtt_msg = create_mqtt_message(msg_data);
    // 发送消息队列
    void *queue = net_app_get_queue();
    if (RT_NULL == queue || module_queue_communication(queue, GATEWAY_MODULE_DEFAULT, NET_APP_MODULE, NET_MQTT_RECV_MESSAGE, (void *)mqtt_msg))
    {
        // 释放 MQTT 消息
        release_mqtt_message(mqtt_msg);
        rt_free(mqtt_msg);
    }
}

/**
 * @brief  创建 net_mqtt 消息队列
 *
 * @return 成功返回0，否则返回非0
 */
int net_mqtt_create_queue(void)
{
    g_net_mqtt_queue = manager_create_queue("net_mqtt", 4, 200, UC_SYSTEM_IPC_FLAG_PRIO);
    if (RT_NULL == g_net_mqtt_queue)
    {
        TRACE_E("manager_create_queue error in %s line %d", __FUNCTION__, __LINE__);
        return 1;
    }
    return 0;
}

/**
 * @brief  获取 net_mqtt 消息队列
 *
 * @return net_mqtt 消息队列
 */
void *net_mqtt_get_queue(void)
{
    return g_net_mqtt_queue;
}

#ifdef MQTT_PACKET_GROUP
/**
 * @brief  创建 mqtt_com 消息队列
 *
 * @return 成功返回0，否则返回非0
 */
int mqtt_com_create_queue(void)
{
    g_mqtt_com_queue = manager_create_queue("mqtt_com", 4, 200, UC_SYSTEM_IPC_FLAG_PRIO);
    if (RT_NULL == g_mqtt_com_queue)
    {
        TRACE_E("manager_create_queue error in %s line %d", __FUNCTION__, __LINE__);
        return 1;
    }
    return 0;
}

/**
 * @brief  获取 mqtt_com 消息队列
 *
 * @return mqtt_com 消息队列
 */
void *mqtt_com_get_queue(void)
{
    return g_mqtt_com_queue;
}
#endif

/**
 * @brief  创建 MQTT 消息
 *
 * @param  data 初始化数据
 * @return
 */
net_mqtt_message_t *create_mqtt_message(const void *data)
{
    // 分配内存
    net_mqtt_message_t *mqtt_msg = rt_malloc(sizeof(net_mqtt_message_t));
    if (RT_NULL == mqtt_msg)
    {
        TRACE_E("%s, %d malloc failed", __FUNCTION__, __LINE__);
        return RT_NULL;
    }
    rt_memset(mqtt_msg, 0, sizeof(net_mqtt_message_t));

    if (RT_NULL != data)
    {
        const message_data *msg_data = (const message_data *)data;
        // 主题
        mqtt_msg->topic = rt_malloc(msg_data->topic_name->lenstring.len + 1);
        if (RT_NULL == mqtt_msg->topic)
        {
            TRACE_E("%s, %d malloc failed", __FUNCTION__, __LINE__);
            rt_free(mqtt_msg);
        }
        rt_memcpy(mqtt_msg->topic, msg_data->topic_name->lenstring.data, msg_data->topic_name->lenstring.len);
        mqtt_msg->topic[msg_data->topic_name->lenstring.len] = '\0';
        // 有效载荷
        mqtt_msg->payload = rt_malloc(msg_data->message->payloadlen + 1);
        if (RT_NULL == mqtt_msg->payload)
        {
            TRACE_E("%s, %d malloc failed", __FUNCTION__, __LINE__);
            rt_free(mqtt_msg->topic);
            rt_free(mqtt_msg);
        }
        rt_memcpy(mqtt_msg->payload, msg_data->message->payload, msg_data->message->payloadlen);
        mqtt_msg->payload[msg_data->message->payloadlen] = '\0';
    }
    return mqtt_msg;
}

/**
 * @brief 释放 MQTT 消息
 *
 * @param msg MQTT 消息
 */
void release_mqtt_message(net_mqtt_message_t *msg)
{
    RT_ASSERT(msg);

    if (RT_NULL != msg->topic)
    {
        rt_free(msg->topic);
        msg->topic = RT_NULL;
    }
    if (RT_NULL != msg->payload)
    {
        rt_free(msg->payload);
        msg->payload = RT_NULL;
    }
}

/**
 * @brief 初始化MQTT参数
 *
 */
static void mqtt_init(void)
{
    gw_base_cfg_t *base_cfg = uc_static_get_gateway_base_cfg();
    gw_mqtt_server_cfg_t *mqtt_cfg = uc_static_get_gateway_mqtt_cfg();

    // 获取静态数据
    rt_memcpy(g_mqtt_config.client_id, mqtt_cfg->client_id, sizeof(mqtt_cfg->client_id));
    rt_memcpy(g_mqtt_config.username, mqtt_cfg->username, sizeof(mqtt_cfg->username));
    rt_memcpy(g_mqtt_config.password, mqtt_cfg->password, sizeof(mqtt_cfg->password));
    // MQTT服务器
    rt_snprintf(g_mqtt_config.url, 64, "tcp://%s:%u", mqtt_cfg->url, mqtt_cfg->port);
    // 订阅主题
    rt_snprintf(g_mqtt_config.sub_topic0, 64, "%s/down/+", base_cfg->company);
    rt_snprintf(g_mqtt_config.sub_topic1, 64, "%s/down/sgw/+", base_cfg->company);
    rt_snprintf(g_mqtt_config.sub_topic2, 64, "%s/down/sgw/%u/+", base_cfg->company, base_cfg->dev_id);
    // 发布主题
    rt_snprintf(g_mqtt_config.pub_topic, 64, "%s/up/sgw/%u", base_cfg->company, base_cfg->dev_id);

    g_mqtt_client.message_handlers[0].qos = mqtt_cfg->qos;
    g_mqtt_client.message_handlers[1].qos = mqtt_cfg->qos;
    g_mqtt_client.message_handlers[2].qos = mqtt_cfg->qos;

    TRACE_I("sub_topic0 %s", g_mqtt_config.sub_topic0);
    TRACE_I("sub_topic1 %s", g_mqtt_config.sub_topic1);
    TRACE_I("sub_topic2 %s", g_mqtt_config.sub_topic2);
}

#ifdef MQTT_PACKET_GROUP

static mqtt_data_manager_t mqtt_data_manager = {0};
static rt_mutex_t g_dynamic_mutex = RT_NULL;

// 初始化 mqtt_data_manager_t
int init_mqtt_data_manager(void)
{
    rt_memset(&mqtt_data_manager, 0, sizeof(mqtt_data_manager));

    // 先按照json格式，将第一个字节加'['
    for (int i = 0; i < MQTT_CACHE_NUM; i++)
    {
        mqtt_data_manager.mqtt_data_cache[i].buf[0] = '[';
        mqtt_data_manager.mqtt_data_cache[i].len = 1;
    }

    g_dynamic_mutex = rt_mutex_create("dmutex", RT_IPC_FLAG_PRIO);
    if (g_dynamic_mutex == RT_NULL)
    {
        TRACE_E("create dynamic mutex failed");
        return -1;
    }

    return 0;
}

#define MQTT_WRITE_FLAG_DEF (0xef)

// 写缓存数据
int write_mqtt_data(char *data, int len)
{
    int full_index;
    int result = 0;
    // 参考长度
    int ref_len = MQTT_SEND_MAX_SIZE;

    // 网络组包，建议长度不超过1500字节。如果单个数据超过1500字节，就用2500最大buffer长度限制组包
    if (len > ref_len)
    {
        TRACE_E("input data len %d too long, suggest len %d", len, ref_len);
        return 1;
    }

    full_index = MQTT_WRITE_FLAG_DEF;

    // TRACE_I("current write index %d, input len %d", mqtt_data_manager.write_index, len);
    rt_mutex_take(g_dynamic_mutex, RT_WAITING_FOREVER);

    if (mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].len + len > ref_len &&
        0 == mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].full_flag)
    { // 缓存无法在缓存一个更大的。
        // TRACE_I("full write_index %d current index len %d", mqtt_data_manager.write_index, mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].len);

        //   记录
        full_index = mqtt_data_manager.write_index;

        // JSON尾巴数据将写‘]’
        mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].buf[mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].len] = ']';
        mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].len++;
        mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].full_flag = 1;

        // 缓存已经满，将新数据丢掉
        if ((mqtt_data_manager.write_index + 1) % MQTT_CACHE_NUM == mqtt_data_manager.read_index)
        {
            result = 1;
            TRACE_E("mqtt_data_manager full, loss data");
            goto _full_notify;
        }
        // 指向下个缓存
        mqtt_data_manager.write_index++;
        if (mqtt_data_manager.write_index >= MQTT_CACHE_NUM)
        {
            // buffer翻转
            mqtt_data_manager.write_index = 0;
        }

        // 数据缓存下个开始，清理下个数据
        mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].buf[0] = '[';
        mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].len = 1;
        mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].full_flag = 0;
    }
    // 当前已经满
    else if (1 == mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].full_flag)
    { // 上次缓存已经满，将新数据丢掉
        full_index = mqtt_data_manager.write_index;

        // TRACE_I("have full write_index %d current index len %d", mqtt_data_manager.write_index, mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].len);

        // 写的下个位置就是读位置，但是读位置是满。则说明缓存已经满了，丢弃数据
        if (((mqtt_data_manager.write_index + 1) % MQTT_CACHE_NUM) == mqtt_data_manager.read_index)
        {

            result = 1;
            TRACE_E("have mqtt_data_manager full, loss new data. w %d r %d", mqtt_data_manager.write_index, mqtt_data_manager.read_index);
            goto _full_notify;
        }
        else
        { // 上次数据满，已经有数据读走。有一个空位置可以缓存数据
            mqtt_data_manager.write_index++;
            if (mqtt_data_manager.write_index >= MQTT_CACHE_NUM)
            {
                // buffer翻转
                mqtt_data_manager.write_index = 0;
            }
            // 清理数据
            mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].buf[0] = '[';
            mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].len = 1;
            mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].full_flag = 0;
        }
    }

    // 当前缓存未满，并且写入非第一个数据，每个数据之间用','隔开
    if (mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].len > 1)
    {
        // TRACE_I("input next");
        mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].buf[mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].len] = ',';
        mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].len++;
    }

    // TRACE_I("cache data index %d len %d", mqtt_data_manager.write_index, mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].len);

    // 当前缓存没有满，可以继续缓存
    rt_memcpy(mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].buf + mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].len, data, len);
    mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].len += len;
    mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].number++;

_full_notify:
    if (full_index != MQTT_WRITE_FLAG_DEF)
    {
        // 发送通知
        int res = rt_mq_send_wait(mqtt_com_get_queue(), &full_index, 4, MODULE_SEND_QUEUE_TIMEOUT);
        if (RT_EOK != res)
        {
            TRACE_E("notify mqtt data %s line %d res=%d", __FUNCTION__, __LINE__, res);
            result = 2;
        }

        // TRACE_W("_full_notify ok");
    }
    rt_mutex_release(g_dynamic_mutex);
    return result;
}
// 获取指定位置数据
int get_the_mqtt_data(char index, char *data, int *len)
{
    int res = 1;
    rt_mutex_take(g_dynamic_mutex, RT_WAITING_FOREVER);
    if (mqtt_data_manager.mqtt_data_cache[index].full_flag == 1 && mqtt_data_manager.mqtt_data_cache[index].len > 1)
    {
        // 读取已经存满的数据
        *len = mqtt_data_manager.mqtt_data_cache[index].len;
        rt_memcpy(data, mqtt_data_manager.mqtt_data_cache[index].buf, *len);
        res = 0;
    }
    rt_mutex_release(g_dynamic_mutex);
    // 读取数据失败
    // TRACE_E("get mqtt data fail");

    return res;
}

// 轮询读缓存数据
int read_mqtt_data(int *index, char **data, int *len)
{
    int result = 1; // 0 : get data, 1 : no data

    // TRACE_I("read index %d write %d", mqtt_data_manager.read_index, mqtt_data_manager.write_index);
    rt_mutex_take(g_dynamic_mutex, RT_WAITING_FOREVER);
    for (int m = mqtt_data_manager.read_index;
         (m <= mqtt_data_manager.write_index ||
          m < MQTT_CACHE_NUM + mqtt_data_manager.write_index && mqtt_data_manager.read_index > mqtt_data_manager.write_index);
         m++)
    {
        int i = m % MQTT_CACHE_NUM;

        if (mqtt_data_manager.mqtt_data_cache[i].full_flag == 1 && mqtt_data_manager.mqtt_data_cache[i].len > 1)
        {
            //   读取已经存满的数据
            *len = mqtt_data_manager.mqtt_data_cache[i].len;
            // rt_memcpy(data, mqtt_data_manager.mqtt_data_cache[i].buf, *len);
            *data = mqtt_data_manager.mqtt_data_cache[i].buf;
            *index = i;

            if (mqtt_data_manager.read_index == mqtt_data_manager.write_index)
            {
                // 位置跳转下个
                mqtt_data_manager.write_index++;

                if (mqtt_data_manager.write_index >= MQTT_CACHE_NUM)
                {
                    // 位置翻转
                    mqtt_data_manager.write_index = 0;
                }
                // 恢复数据
                mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].buf[0] = '[';
                mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].len = 1;
                mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].full_flag = 0;
            }

            result = 0;

            break;
        }
        else if (mqtt_data_manager.mqtt_data_cache[i].full_flag == 0 && /*数据未满，强制将已经有的数据拷贝走*/
                 mqtt_data_manager.mqtt_data_cache[i].len > 1 &&
                 mqtt_data_manager.read_index == mqtt_data_manager.write_index)
        {
            //  json完整数据
            mqtt_data_manager.mqtt_data_cache[i].buf[mqtt_data_manager.mqtt_data_cache[i].len] = ']';
            mqtt_data_manager.mqtt_data_cache[i].len++;
            mqtt_data_manager.mqtt_data_cache[i].full_flag = 1;

            // 位置跳转下个
            mqtt_data_manager.write_index++;

            if (mqtt_data_manager.write_index >= MQTT_CACHE_NUM)
            {
                // 位置翻转
                mqtt_data_manager.write_index = 0;
            }

            // 恢复数据
            mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].buf[0] = '[';
            mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].len = 1;
            mqtt_data_manager.mqtt_data_cache[mqtt_data_manager.write_index].full_flag = 0;

            // 读取数据
            *index = i;
            *len = mqtt_data_manager.mqtt_data_cache[i].len;
            // rt_memcpy(data, mqtt_data_manager.mqtt_data_cache[i].buf, *len);
            *data = mqtt_data_manager.mqtt_data_cache[i].buf;

            result = 0;
            break;
        }
        // TRACE_W("current cache is empty");
    }

    rt_mutex_release(g_dynamic_mutex);

    // TRACE_I("result %d after read index %d write %d", result, mqtt_data_manager.read_index, mqtt_data_manager.write_index);
    return result;
}

// 清理指定数据
int clear_mqtt_data(int index)
{
    int res = 1;

    if (index >= MQTT_CACHE_NUM)
    {
        TRACE_E("index %d is out of range", index);
        return 1;
    }

    rt_mutex_take(g_dynamic_mutex, RT_WAITING_FOREVER);

    if (index == mqtt_data_manager.read_index)
    {
        res = 0;
    }
    else
    {
        TRACE_E("clear index %d, read_index %d", index, mqtt_data_manager.read_index);
    }

    mqtt_data_manager.mqtt_data_cache[index].full_flag = 0;
    mqtt_data_manager.mqtt_data_cache[index].len = 1;
    mqtt_data_manager.mqtt_data_cache[index].number = 0;

    // if (mqtt_data_manager.read_index != mqtt_data_manager.write_index)
    {
        mqtt_data_manager.read_index++;
        if (mqtt_data_manager.read_index >= MQTT_CACHE_NUM)
        {
            mqtt_data_manager.read_index = 0;
        }
    }
    rt_mutex_release(g_dynamic_mutex);
    return res;
}

unsigned int mqtt_push_checkout_timeout = 0;

void mqtt_push_checkout(void)
{
    if (0 != mqtt_push_checkout_timeout)
    {
        // rt_base_t level = rt_hw_interrupt_disable();
        unsigned int the_tick = rt_tick_get();

        if (((the_tick > mqtt_push_checkout_timeout) && (the_tick - mqtt_push_checkout_timeout > 30000)) ||
            ((the_tick < mqtt_push_checkout_timeout) && (mqtt_push_checkout_timeout - the_tick > 30000)))
        {
            rt_kprintf("mqtt task anbormal.now reboot. check time %d current tick %d", mqtt_push_checkout_timeout, the_tick);
            // rt_thread_delay(100);
            //  rt_hw_cpu_reset();
            //  while (1)
            //      ;
            rt_thread_mdelay(100);
            extern void watchdog_reboot(void);
            watchdog_reboot();
        }
        // rt_hw_interrupt_enable(level);
    }
}

static void mqtt_com_entry(void *para)
{
    int index;
    int rc;
    int ret;
    int len;
    int timeout;
    char *buffer;
    char *topic;
    int topic_len;
    struct mqtt_client *mqtt_res = (struct mqtt_client *)para;
    int all_count = 0;

    TRACE_I("mqtt_com_entry");

    topic = rt_malloc(80);
    RT_ASSERT(RT_NULL != topic);

    rt_snprintf(topic, 80, "%s/", g_mqtt_config.pub_topic);
    topic_len = rt_strlen(topic);

    // buffer = rt_malloc(MQTT_SEND_MAX_SIZE);
    // RT_ASSERT(RT_NULL != buffer);

    timeout = MQTT_CACHE_WAIT_TIMEOUT;

    while (1)
    {
        rc = rt_mq_recv(g_mqtt_com_queue, &index, 4, timeout);
        if ((-RT_ETIMEOUT == rc || QUEUE_EOK == rc) && MQTT_STATE_ONLINE == manager_get_mqtt_state())
        {
            // if (delay_flag)
            //{
            //     TRACE_W("wait 500ms");
            //     delay_flag = 0;
            //     rt_thread_delay(500);
            // }
            //  int count = 0;
            //   轮询数据

            if (0 == read_mqtt_data(&index, &buffer, &len))
            {

                // 发送数据
                rt_snprintf(topic + topic_len, 80 - topic_len, "%u", g_mqtt_publish_num++);
                ret = paho_mqtt_publish(&g_mqtt_client, mqtt_res->message_handlers[0].qos, topic, buffer, len);

                // 清理数据
                if (ret == 0)
                {

                    all_count += mqtt_data_manager.mqtt_data_cache[index].number;
                    clear_mqtt_data(index);
                }
                send_led_event(LED_MQTT, LED_E_BLINK);

                TRACE_I("task %s MQTT published %s data len %d index %d allnum %d",
                        (rc == -RT_ETIMEOUT ? "timeout" : "ok"),
                        (ret == 0 ? "succ" : "failed"),
                        len,
                        index,
                        all_count);
            }
            timeout = MQTT_CACHE_TIMEOUT;
        }
        else
        {
            timeout = MQTT_CACHE_WAIT_TIMEOUT;
            // delay_flag = 1;
        }

        mqtt_push_checkout_timeout = rt_tick_get();
    }
}
#endif

/**
 * @brief  启动 MQTT
 *
 * @return 成功返回0，失败返回非0
 */
int mqtt_start(void)
{
    if (g_mqtt_started == 1)
    {
        TRACE_I("warning: mqtt client is connected");
        return -1;
    }
    if (g_mqtt_state != MQTT_STATE_OFFLINE)
    {
        TRACE_W("warning: try again after offline");
        return -2;
    }

    // init MQTT client
    mqtt_init();

    // init condata param by using MQTTPacket_connectData_initializer
    MQTTPacket_connectData condata = MQTTPacket_connectData_initializer;

    // config MQTT context param
    g_mqtt_client.isconnected = 0;
    g_mqtt_client.uri = g_mqtt_config.url;

    // config connect param
    memcpy(&g_mqtt_client.condata, &condata, sizeof(condata));

    g_mqtt_client.condata.clientID.cstring = g_mqtt_config.client_id;
    if (rt_strlen(g_mqtt_config.username) > 0)
    {
        g_mqtt_client.condata.username.cstring = g_mqtt_config.username;
    }
    if (rt_strlen(g_mqtt_config.password) > 0)
    {
        g_mqtt_client.condata.password.cstring = g_mqtt_config.password;
    }
    g_mqtt_client.condata.keepAliveInterval = MQTT_KEEP_ALIVE_INTERVAL;
    g_mqtt_client.condata.cleansession = MQTT_CLEAN_SESSION;

    // malloc buffer
    g_mqtt_client.buf_size = MQTT_BUFFER_SIZE;
    g_mqtt_client.buf = rt_malloc(g_mqtt_client.buf_size);
    if (RT_NULL == g_mqtt_client.buf)
    {
        TRACE_E("error: no memory for MQTT client buffer");
        return 1;
    }
    rt_memset(g_mqtt_client.buf, 0, g_mqtt_client.buf_size);

    // allocate read buffer
    g_mqtt_client.readbuf_size = MQTT_READ_BUFFER_SIZE;
    g_mqtt_client.readbuf = rt_malloc(g_mqtt_client.readbuf_size);
    if (RT_NULL == g_mqtt_client.readbuf)
    {
        TRACE_E("error: no memory for MQTT client read buffer");
        rt_free(g_mqtt_client.buf);
        return 1;
    }
    rt_memset(g_mqtt_client.readbuf, 0, g_mqtt_client.readbuf_size);

    // set event callback function
    g_mqtt_client.connect_callback = mqtt_connect_callback;
    g_mqtt_client.online_callback = mqtt_online_callback;
    g_mqtt_client.offline_callback = mqtt_offline_callback;

    // set subscribe table and event callback
    g_mqtt_client.message_handlers[0].topicFilter = rt_strdup(g_mqtt_config.sub_topic0);
    g_mqtt_client.message_handlers[1].topicFilter = rt_strdup(g_mqtt_config.sub_topic1);
    g_mqtt_client.message_handlers[2].topicFilter = rt_strdup(g_mqtt_config.sub_topic2);
    g_mqtt_client.message_handlers[0].callback = mqtt_sub_callback;
    g_mqtt_client.message_handlers[1].callback = mqtt_sub_callback;
    g_mqtt_client.message_handlers[2].callback = mqtt_sub_callback;
    // g_mqtt_client.message_handlers[0].qos = QOS0;
    // g_mqtt_client.message_handlers[1].qos = QOS0;
    // g_mqtt_client.message_handlers[2].qos = QOS0;

    int value;
    int u16Value;
    value = 5;
    paho_mqtt_control(&g_mqtt_client, MQTT_CTRL_SET_CONN_TIMEO, &value);
    value = 5;
    paho_mqtt_control(&g_mqtt_client, MQTT_CTRL_SET_MSG_TIMEO, &value);
    value = 5;
    paho_mqtt_control(&g_mqtt_client, MQTT_CTRL_SET_RECONN_INTERVAL, &value);
    value = 11;
    paho_mqtt_control(&g_mqtt_client, MQTT_CTRL_SET_KEEPALIVE_INTERVAL, &value);
    u16Value = 3;
    paho_mqtt_control(&g_mqtt_client, MQTT_CTRL_SET_KEEPALIVE_COUNT, &u16Value);

    // run mqtt client
    paho_mqtt_start(&g_mqtt_client, MQTT_THREAD_STACK_SIZE, MQTT_THREAD_PRIORITY);
    g_mqtt_started = 1;

#ifdef MQTT_PACKET_GROUP
    // mqtt上行组包线程
    void *mqtt_com_handle = RT_NULL;

    mqtt_com_create_queue();

    init_mqtt_data_manager();
    // = rt_mutex_create("mqttmut", RT_IPC_FLAG_FIFO);
    // RT_ASSERT(RT_NULL != g_mqtt_mutex);
    if (0 != manager_thread_create(&mqtt_com_handle, "mqtt_com", mqtt_com_entry, &g_mqtt_client, 512, MQTT_THREAD_PRIORITY, 10))
    {
        TRACE_E("mqtt_com task create error");
    }

    rt_thread_startup((rt_thread_t)mqtt_com_handle);
#endif

    return 0;
}

/**
 * @brief  终止 MQTT
 *
 * @return 成功返回0，失败返回非0
 */
int mqtt_stop(void)
{
    if (g_mqtt_started == 0)
    {
        TRACE_W("warning: mqtt client is not connected");
        return -1;
    }
    g_mqtt_started = 0;
    TRACE_I("info: waiting to stop MQTT");

    return paho_mqtt_stop(&g_mqtt_client);
}

/**
 * @brief  MQTT 订阅主题
 *
 * @param  argc 参数个数
 * @param  argv 命令行参数
 * @return 成功返回0，失败返回非0
 */
int mqtt_subscribe(int argc, char **argv)
{
    if (argc != 2)
    {
        TRACE_E("error: mqtt_subscribe [topic]");
        return -1;
    }
    if (g_mqtt_started == 0)
    {
        TRACE_W("warning: mqtt client is not connected");
        return -1;
    }
    return paho_mqtt_subscribe(&g_mqtt_client, QOS1, argv[1], mqtt_sub_callback);
}

/**
 * @brief  MQTT 取消订阅
 *
 * @param  argc 参数个数
 * @param  argv 命令行参数
 * @return 成功返回0，失败返回非0
 */
int mqtt_unsubscribe(int argc, char **argv)
{
    if (argc != 2)
    {
        TRACE_E("error: mqtt_unsubscribe [topic]");
        return -1;
    }
    if (g_mqtt_started == 0)
    {
        TRACE_W("warning: mqtt client is not connected");
        return -1;
    }
    return paho_mqtt_unsubscribe(&g_mqtt_client, argv[1]);
}

/**
 * @brief  MQTT 发布消息
 *
 * @param  argc 参数个数
 * @param  argv 命令行参数
 * @return 成功返回0，失败返回非0
 */
int mqtt_publish(int argc, char **argv)
{
    int rc = 0;
    if (g_mqtt_started == 0)
    {
        TRACE_W("warning: mqtt client is not connected");
        return -1;
    }
    if (argc == 2)
    {
        char topic[80];
        rt_snprintf(topic, 80, "%s/%u", g_mqtt_config.pub_topic, g_mqtt_publish_num++);
        rc = paho_mqtt_publish(&g_mqtt_client, QOS1, topic, argv[1], strlen(argv[1]));
    }
    else if (argc == 3)
    {
        rc = paho_mqtt_publish(&g_mqtt_client, QOS1, argv[1], argv[2], strlen(argv[2]));
    }
    else
    {
        TRACE_E("error: mqtt_publish <topic> [message]");
        return -2;
    }
    return rc;
}

#ifdef FINSH_USING_MSH
MSH_CMD_EXPORT(mqtt_start, startup mqtt client);
MSH_CMD_EXPORT(mqtt_stop, stop mqtt client);
MSH_CMD_EXPORT(mqtt_subscribe, mqtt subscribe topic);
MSH_CMD_EXPORT(mqtt_unsubscribe, mqtt unsubscribe topic);
MSH_CMD_EXPORT(mqtt_publish, mqtt publish message to specified topic);
#endif /* FINSH_USING_MSH */
#endif
